package com.zyx.flinkdemo.dataset.stateprocessor;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.state.api.ExistingSavepoint;
import org.apache.flink.state.api.Savepoint;

import java.io.IOException;

/**
 * @author zyx
 * @since 2021/5/25 23:11
 * desc: StateProcessor读取实例
 */
public class StateProcessorReadDemo {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 载入保存点
        ExistingSavepoint savepoint = Savepoint.load(env,
                "hdfs://ns01/test/checkpoint_test/savepoint-1", new HashMapStateBackend());
        // 从保存点读取运算符
        DataSource<Integer> listState = savepoint.readListState("uid1", "state1", Types.INT);
        listState.print();
    }
}
